package com.my.study.batch;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 模拟合并请求
 * <p>
 * 6个库存，10个用户同时进行抢购
 * </p>
 *
 * @author Carlos
 * @version 1.0.0
 * @date 2023/4/2 20:26
 */
public class MergeRequestDemo {

    private Integer stock = 6;

    private BlockingQueue<ResultPromise> queue = new LinkedBlockingQueue<>(10);

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newCachedThreadPool();
        MergeRequestDemo demo = new MergeRequestDemo();
        demo.mergeJob();
        Thread.sleep(2000);
        
        List<Future<Result>> resultFutures = new ArrayList<>();

        CountDownLatch countDownLatch = new CountDownLatch(10);
        for (int i = 0; i < 10; i++) {
            final long orderId = i + 100;
            final long userId = i;

            Future<Result> future = executor.submit(() -> {
                countDownLatch.countDown();
                countDownLatch.await(1000, TimeUnit.SECONDS);
                return demo.operate(new UserRequest(orderId, userId, 1));
            });
            resultFutures.add(future);
        }
        
        resultFutures.forEach(resultFuture -> {
            try {
                Result result = resultFuture.get(300, TimeUnit.MILLISECONDS);
                System.out.println(Thread.currentThread().getName() + " 客户端请求响应：" + result);
            } catch (InterruptedException | TimeoutException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 用户库存扣减
     *
     * @param request: 参数
     * @return Result
     * @author Carlos
     * @date 2023/4/2 20:49
     */
    public Result operate(UserRequest request) throws InterruptedException {
        ResultPromise resultPromise = new ResultPromise(request);
        
        // 这里不是为了加锁，只是用做等待唤醒而已！！！
        synchronized (resultPromise) {
            boolean offer = queue.offer(resultPromise, 100, TimeUnit.MILLISECONDS);
            if (!offer) {
                return new Result(false, "系统繁忙");
            }
            try {
                resultPromise.wait(200);
            } catch (InterruptedException e) {
                return new Result(false, "等待超时");
            }
        }

        return resultPromise.getResult();
    }
    
    public void mergeJob() {
        new Thread(() -> {
            List<ResultPromise> list = new ArrayList<>();
            while (true) {
                // 如果是空的，休眠10ms
                if (queue.isEmpty()) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(10);
                        continue;
                    } catch (InterruptedException e) {
                        // 日志记录
                    }
                }
                
                while (queue.peek() != null) {
                    list.add(queue.poll());
                }

                System.out.println(Thread.currentThread().getName() + "合并扣减库存：" + list);

                int sum = list.stream().mapToInt(e -> e.getUserRequest().getCount()).sum();
                
                if (sum <= stock) {
                    stock -= sum;
                    // 通知所有用户
                    list.forEach(resultPromise -> {
                        resultPromise.setResult(new Result(true, "ok"));
                        synchronized (resultPromise) {
                            resultPromise.notify();
                        }
                    });
                    // 继续轮询
                    continue;
                }

                for (ResultPromise resultPromise : list) {
                    int count = resultPromise.getUserRequest().getCount();
                    if (count <= stock) {
                        stock -= count;
                        resultPromise.setResult(new Result(true, "ok"));
                    }
                    else {
                        resultPromise.setResult(new Result(false, "库存不够"));
                    }
                    synchronized (resultPromise) {
                        resultPromise.notify();
                    }
                }
            }
        }, "mergeJob").start();
        
    }

}

@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
class ResultPromise {
    private UserRequest userRequest;
    private Result result;

    public ResultPromise(UserRequest userRequest) {
        this.userRequest = userRequest;
    }
}


@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
class UserRequest {
    private Long orderId;
    private Long userId;
    private Integer count;
}

@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
class Result {
    private Boolean isSuccess;
    private String msg;

}



